import org.apache.hadoop.hbase.{HBaseConfiguration, TableName}
import org.apache.hadoop.hbase.client.{ConnectionFactory, Scan}
import org.apache.hadoop.hbase.mapreduce.TableInputFormat
import org.apache.hadoop.hbase.protobuf.ProtobufUtil
import org.apache.hadoop.hbase.util.Bytes
import org.apache.spark.graphx.PartitionStrategy.EdgePartition2D
import org.apache.spark.rdd.RDD
import org.apache.spark.graphx._
import org.apache.spark.{SparkConf, SparkContext}

import java.util.Base64
import org.json4s._
import org.json4s.jackson.JsonMethods._

import java.nio.ByteBuffer
import java.nio.channels.FileChannel
import java.nio.file.{Paths, StandardOpenOption}
//import com.google.protobuf.LiteralByteString

// 在IDEA的命令行中运行sbt run后运行一段时间会报内存不足的错误,尝试使用命令:
// set JAVA_OPTS=-Xmx4G -Xms1G && sbt run
// 但是IDEA报错：
// --------------------------------------------------------------
// 所在位置 行:1 字符: 29
//+ set JAVA_OPTS=-Xmx4G -Xms1G && sbt run
//+                             ~~
//标记“&&”不是此版本中的有效语句分隔符。
//    + CategoryInfo          : ParserError: (:) [], ParentContainsErrorRecordException
//    + FullyQualifiedErrorId : InvalidEndOfLine
// --------------------------------------------------------------
// 最后只能转移到cmd中执行。


object HBaseNewAPIReader {
  private val grid_cols = 100
  private val grid_rows = 100
  private var map_grid = Array.fill[Byte](grid_cols, grid_rows)(0)
  /**
   * 从hbase读取数据：[(row-key-0, accessible[100]), ...., (row-key-99, accessible[100])]
   * @param sc 一个SparkContext
   * @return row-key开始的位置的1度乘以1度范围内的是否可通行的数据，共100条，格式如：
   *         [(row-key-0, accessible[100]), ...., (row-key-99, accessible[100])]
   *         数组中accessible[100]是0.1x0.1度的数据（即一个格子是0.01度，10乘以10）
   */
  private def getData(coordX: Int, corrdY: Int, sc: SparkContext):  Array[(String, List[Int])] = {
    // 3. 配置HBase
    val conf = HBaseConfiguration.create()
    conf.set("hbase.zookeeper.quorum", "192.168.1.208:2181,192.168.1.208:2182,192.168.1.208:2183")
    //conf.set("hbase.zookeeper.quorum", "192.168.1.208")
    //conf.set("hbase.zookeeper.property.clientPort", "7077")

    // 4. 设置要读取的表
    val tableName = "mapdata"
    conf.set(TableInputFormat.INPUT_TABLE, tableName)

    // 5. 配置Scan对象（关键部分）
    // 现在我们读取119.0开始的一个经度宽的数据：(119.0, 23.0) -> (120.0, 24.0)

    val scan = new Scan()
    val (beginX, endX) = (f"$coordX%08x00000000", f"${coordX+100}%08x00000000")
    scan.withStartRow(Bytes.toBytes(beginX)) // 起始行："000074cc00000000"
    scan.withStopRow(Bytes.toBytes(endX))   // 结束行: "0000753000000000"
    scan.addColumn(Bytes.toBytes("route"), Bytes.toBytes("accessible")) // 指定列族和列
    scan.setCacheBlocks(false) // 不缓存块
    //scan.setLimit(10).setCaching(10)  // 测试！限制只返回10条只缓存10条！

    // 6. 将Scan对象转换为Base64字符串
    val scanToString = Base64.getEncoder.encodeToString(ProtobufUtil.toScan(scan).toByteArray)
    conf.set(TableInputFormat.SCAN, scanToString)

    // 7. 使用newAPIHadoopRDD读取数据
    val hbaseRDD = sc.newAPIHadoopRDD(
      conf,
      classOf[TableInputFormat],
      classOf[org.apache.hadoop.hbase.io.ImmutableBytesWritable],
      classOf[org.apache.hadoop.hbase.client.Result]
    )

    // 8. 处理结果，测试时临时打印一下，这个Action不必要。
    val resultCount = hbaseRDD.count()
    println(s"[*****] before filter: We got $resultCount records [*****]")

    // 9. 转换为可读格式并打印前10条
    val results = hbaseRDD
      .filter {
        // hbase需要重新设计row-key以达到一次scan返回所有需要数据的目的，这里采用filter是权宜之计
        case (_, result) =>
          val rowKey = Bytes.toString(result.getRow)
          val subStr = rowKey.substring(8, 16)
          val intValue = Integer.parseInt(subStr, 16)
          intValue >= corrdY && intValue < corrdY+100  // 半区间[start, end)
      }
      .map { case(_, result) =>
        val rowKey = Bytes.toString(result.getRow)
        val arrayBytes = result.getValue(Bytes.toBytes("route"), Bytes.toBytes("accessible"))
        // arrayBytes类似这样b'[1,1,1,1,1,1,1,...,1,1,1]'，它不会以0结尾。
        val arrayJson = if (arrayBytes!=null) Bytes.toString(arrayBytes) else "[]"
        // 转成字符串"[1,1,1,1,1,1,1,...,1,1,1]"
        // 使用json4s解析JSON数组
        implicit val formats = DefaultFormats
        val array = parse(arrayJson).extract[List[Int]] // 根据实际数组元素类型调整
        (rowKey, array)
      }.collect // 收集到一个drive中，方便后面创建图（这一步未来要优化，收集数据很浪费时间）

    val result2Count = results.length
    println(s"[*****] after filter: We got $result2Count records [*****]")

    // 6. 测试，打印数组
//    println("===== Records read from HBase =====")
//    results.foreach { case (rowKey, array) =>
//      println(s"RowKey: $rowKey")
//      println(s"Array: ${array.mkString("[", ", ", "]")}")
//      println("-----------------------------")
//    }
    results
  }

  /**
   * 将hbase原始数据转为100x100的二维数据。在经过合理调整hbase的rowkey之后，可能不再需要这个函数（甚至不在需要RDD.collect）
   *
   * @param ia [("000074cc00002c24", accessible[100]), ...., ("0000752600002c7e", accessible[100])]
   * @return 返回100x100的二维数组，每个数组的下标[i][j]对应地图上0.01度的位置
   */
  private def hbaseDataToGridData(coordX: Int, coordY: Int, ia: Array[(String, List[Int])]): Unit = {
    for ((rowKey, l100) <- ia) {
      val i: Int = Integer.parseInt(rowKey.substring(0, 8), 16) - coordX
      val j: Int = Integer.parseInt(rowKey.substring(8, 16), 16) - coordY
      for (x <- 0 to 9) {
        for (y <- 0 to 9) {
          map_grid(i + x)(j + y) = l100(10 * x + y).toByte
        }
      }
    }
  }

  private def printAndSaveGrid(s: Array[Array[Byte]]): Unit = {
    // 测试，打印数组
    println(s"lines: ${s.length}")
    println(s"cols: ${s(0).length}")
    println("finally we get a grid:")
    s.foreach { row =>
      println(row.mkString(" "))
    }
    // 测试，保存到一个文件。
    val buffer = ByteBuffer.allocate(100 * 100)
    for (row <- s; elem <- row) {
      buffer.put(elem)
    }
    val path = Paths.get(".\\target\\output.bin")
    val channel = FileChannel.open(path, StandardOpenOption.CREATE, StandardOpenOption.WRITE)
    buffer.flip() // 准备读取
    channel.write(buffer)
    channel.close()
  }

  /**
   * 单源最短路径 (SSSP)
   * @param graph 图结构 顶点attr是网格坐标(i,j),边attr是权重(固定是1)
   * @param sourceId 起点ID
   * @return
   */
  def shortestPath(graph: Graph[(Int, Int), Int], sourceId: VertexId): Graph[(Double, List[VertexId]), Int] = {
    // 顶点map操作 VE [(Int, Int)] => VE [(Double, List[VertexId])],即(距离, 经过节点列表)
    val initialGraph: Graph[(Double, List[VertexId]), Int] = graph.mapVertices {
      case (id, _) =>
        if (id == sourceId)
          (0.0, List(id))
        else
          (Double.PositiveInfinity, List())
    }

    // 初始消息：(Double.PositiveInfinity:距离, List[VertexId]())
    val sssp = initialGraph.pregel((Double.PositiveInfinity, List[VertexId]()))(
      // 顶点处理函数：收到消息后如果消息中的距离小于当前值，则将消息中的距离代替当前值。
      (_, dist, newDist) => if (dist._1 < newDist._1) dist else newDist,

      // 发送消息函数
      triplet => {
        if (triplet.srcAttr._1 + triplet.attr < triplet.dstAttr._1) {
          Iterator((triplet.dstId, (triplet.srcAttr._1 + triplet.attr, triplet.srcAttr._2 :+ triplet.dstId)))
        } else {
          Iterator.empty
        }
      },

      // 合并消息函数：取距离小者
      (a, b) => if (a._1 < b._1) a else b
    )

    sssp
  }

  // 查询任意两点间最短路径
  def getShortestPath(graph: Graph[(Int, Int), Int],
                      start: (Int, Int),
                      end: (Int, Int)): Option[List[(Int, Int)]] = {
    val startId = start._1 * 100 + start._2
    val endId = end._1 * 100 + end._2

    // 检查起点和终点是否可通行
    if (map_grid(start._1)(start._2) != 1 || map_grid(end._1)(end._2) != 1) {
      return None
    }

    val shortestPaths = shortestPath(graph, startId)

    shortestPaths.vertices.filter(_._1 == endId)
      .first()._2 match {
      case (distance, path) if java.lang.Double.isFinite(distance) =>
        Some(path.map(id => ((id / 100).toInt, (id % 100).toInt)))
      case _ => None
    }
  }

  def main(args: Array[String]): Unit = {
    // 1. 创建Spark配置
    val sparkConf = new SparkConf()
      .setAppName("HBaseNewAPIReader")
      .setIfMissing("spark.master", "local[*]") // 开发环境默认值
      .set("spark.driver.memory", "2g")  // 尝试处理错误（然并卵）: System memory 259522560 must be at least 471859200. Please increase heap size using the --driver-memory
    // 2. 创建SparkContext
    val sc = new SparkContext(sparkConf)
    try {
      val (lan, lat) = (119.0, 23.0)
      println(s"[*****] init (lan,lat)=($lan, $lat) [*****]")

      val coordX = Math.round((lan + 180)*100).toInt
      val coordY = Math.round((lat + 90)*100).toInt  // (74cc, 2c24)
      println(s"[*****] init (x,y)=(${coordX.toHexString}, ${coordY.toHexString}) [*****]")

      val r = getData(coordX, coordY, sc)
      hbaseDataToGridData(coordX, coordY, r)
      printAndSaveGrid(map_grid)

      // 将网格grid转为graphx图结构
      // 创建顶点RDD: grid(i)(j) == 1是可通行，0是不可通行。取所有可通行的网格，将id置为(i*100+j)，顶点将属性设置为(i,j)
      val vertices: RDD[(VertexId, (Int, Int))] = sc.parallelize(
        (0 until 100).flatMap { i =>
          (0 until 100).filter { j => map_grid(i)(j) == 1 }
            .map { j => ((i * 100 +j).toLong, (i, j)) } // 使用i*100+j作为唯一ID
        }
      )
      // 创建边RDD(4连通): Edge[Int]表示边属性是整形数，这里存放的是权重，为方便测试当前固定设置为1.
      val edges: RDD[Edge[Int]] = vertices.flatMap {
        case (id, (i, j)) =>
          Seq(
            // 只生成向右和向下的边，避免重复
            (i+1, j), // right
            (i, j+1) // up
          ).collect {
            case (ni, nj)  if ni >= 0 && ni < 100 && nj >= 0 && nj < 100 && map_grid(ni)(nj) == 1 =>
              Edge(id, ni * 100 + nj, 1) // (srcId, destId, 权重)。测试权重统一为1. (ni * 100 + nj)是这条边连接的目标节点。
          }
      }
      // deepseek推荐的优化(待研究)
      // (1)自定义分区器，将相邻顶点分配到相同分区
      // class VertexRangePartitioner(partitions: Int) extends Partitioner
      // 构建顶点 RDD 并应用自定义分区器
      // val partitionedVertices = vertices.partitionBy(new VertexRangePartitioner(2))
      // 构建边 RDD（边会自动跟随顶点分区）
      // val edges: RDD[Edge[String]] = ... // 你的边数据
      // val graph = Graph(partitionedVertices, edges)
      //
      // (2)自定义分区策略：按顶点 ID 范围分配，使相邻顶点更可能在同一分区
      // spark2：object CustomPartitionStrategy extends PartitionStrategy，spark3已经不允许这样做
      // 在 Spark 3.x 版本中，PartitionStrategy 是一个 sealed trait，它已经内置了 5 种实现（如 RandomVertexCut、EdgePartition2D 等），不允许用户自定义新的实现。
      // val partitionedGraph = graph.partitionBy(EdgePartition2D)

      // 构建图
      val graph = Graph(vertices, edges)
      val path = getShortestPath(graph, (0,0), (99,99))
      path match {
        case Some(p) =>
          println(s"Path found(${p.length} steps):")
          println(p.mkString(" -> "))
        case None =>
          println("No valid path exists.")
      }
    } finally {
      // 10. 关闭SparkContext
      sc.stop()
    }
  }
}

// 下面这几行会报错，是因为序列化的问题，在下方使用map进行数据格式转换之后，可以正常打印数组了：
//      hbaseRDD.take(10).foreach { case (_, result) =>
//        val rowKey = Bytes.toString(result.getRow)
//        println(s"RowKey: $rowKey")
//        //val value = Bytes.toString(result.getValue(Bytes.toBytes("route"), Bytes.toBytes("accessible")))
//        //println(s"RowKey: $rowKey, Value: $value")
//      }